package epoch.db.cmd;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ThreadFactory;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import epoch.db.cache.EntityCache;
import epoch.db.event.DBEvent;
import epoch.db.event.EntityEventFactory;
import epoch.game.core.common.Loggers;

/**
 * 数据缓存管理器
 * 
 * @author will
 * @date 2016年8月10日 上午10:40:19
 */
public class DBCommandCache implements EventHandler<DBEvent>{
	// ringbufferSize 必须是2的次方
	public static int MAX_QUEUE_SIZE = 1024;
	//线程池
	ThreadFactory threadFactory = null;
	//数据操作执行器
	public Disruptor<DBEvent> dbCacheDisruptor = null;
	//缓存
	private Map<String, EntityCache> caches = new HashMap<>();
	private IDBCommandExecutor cmdExecutor;
	private static DBCommandCache me = null;
	
	public static DBCommandCache getMe() {
		return me;
	}
	
	public static DBCommandCache buildInstance(int ringBufferSize,IDBCommandExecutor cmdExecutor){
		me = new DBCommandCache(ringBufferSize, cmdExecutor);
		return me;
	}
	
	public static DBCommandCache buildInstance(int ringBufferSize, EventHandler<DBEvent> eventHandler, IDBCommandExecutor cmdExecutor){
		me = new DBCommandCache(ringBufferSize,cmdExecutor);
		return me;
	}
		
	private DBCommandCache(int ringBufferSize, IDBCommandExecutor cmdExecutor){
		init(ringBufferSize, this, cmdExecutor);
	}
	
	public DBCommandCache(int ringBufferSize, EventHandler<DBEvent> eventHandler, IDBCommandExecutor cmdExecutor){
		init(ringBufferSize,eventHandler, cmdExecutor);
	}
	
	public DBCommandCache(EventHandler<DBEvent> eventHandler){
		init(MAX_QUEUE_SIZE, eventHandler, cmdExecutor);
	}
	
	@SuppressWarnings("unchecked")
	public void init(int ringBufferSize, EventHandler<? super DBEvent> eventHandler, IDBCommandExecutor cmdExecutor) {
		this.cmdExecutor = cmdExecutor;
//		threadFactory = new DefaultThreadFactory("DB-OPERATE");
		threadFactory = (r)-> {
				return null;
		};
		dbCacheDisruptor = new Disruptor<>(new EntityEventFactory(), ringBufferSize, threadFactory, ProducerType.SINGLE, new BlockingWaitStrategy());
		dbCacheDisruptor.handleEventsWith(eventHandler);
	}

	public void start() {
		dbCacheDisruptor.start();
	}
	
	public void stop(){
		dbCacheDisruptor.shutdown();
	}

	/**
	 * 添加操作
	 * @param dbCmd
	 */
	public void addDBCommand(DBCommand dbCmd){
		RingBuffer<DBEvent> ringBuffer = dbCacheDisruptor.getRingBuffer();
		long sequence = ringBuffer.next();
		try {
			DBEvent dbEvent = ringBuffer.get(sequence);
			dbEvent.set(dbCmd);
		} finally {
			//可选择什么时候publish
			ringBuffer.publish(sequence);
		}
	}

	@Override
	public void onEvent(DBEvent dbEvent, long sequence, boolean endOfBatch) throws Exception {
		DBCommand dbCmd = dbEvent.dbCmd;
		Loggers.DB_LOG.debug("handleEvent:optType={}, sequence={},endOfBatch={},thread={}, dbCmd.entityName={}"
				,dbEvent.dbCmd.optType,sequence,endOfBatch,Thread.currentThread().getName()
				,((EntityCommand)dbCmd).entity.tableName());
		//每个线程去处理自己对应的数据
		
		EntityCache cache = caches.get(dbCmd.cacheKey);
		if(cache==null){
			cache = new EntityCache();
			caches.put(dbCmd.cacheKey, cache);
		}
		cache.addDbTask(dbCmd);
		dbCmd = cache.poll();
		if(dbCmd!=null){
			//dbCmd.execute();
		}
	}

}
